package services

import (
	"sync"
	"time"

	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/apis"
	app_process "codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/models/appprocess"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/redisserver"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/services/event"

	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/common"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/models"
	server_session "codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/models/serversession"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/services/stragegy"
	"codehub-g.huawei.com/videocloud/mediaprocesscenter/application-gateway/pkg/utils/log"
)

const (
	roundThreshold           = 60
	judgeTime                = 300
	NoAvailableServerSession = "there is no available process"
)

type ServerSessionDispatcher struct {
	roundCount int32
}

func (d *ServerSessionDispatcher) Work(stopCh chan struct{}) {
	d.roundCount = 0
	go d.work(stopCh)
}

func (d *ServerSessionDispatcher) work(stopCh chan struct{}) {
	ticker := time.NewTicker(judgeTime * time.Second)
	for {
		select {
		case <-stopCh:
			ticker.Stop()
			return
		case <-ticker.C:
			d.monitor()
		default:
			d.executeOneDispatchRound()
		}
	}
}

func (d *ServerSessionDispatcher) executeOneDispatchRound() {
	// 获取所有server session
	serverSessionDao := server_session.NewServerSessionDao(models.MySqlOrm)
	// 先到先处理，按创建时间升序获取
	sssDB, err := serverSessionDao.ListByState1(common.ServerSessionStateCreating, common.ASCSort, 0, 10000)
	if err != nil {
		log.RunLogger.Errorf("[dispatch] failed to get server sessions with "+" for %v", err)
		return
	}

	log.RunLogger.Infof("[dispatch] start to dispatch %d server session", len(*sssDB))
	wg := sync.WaitGroup{}
	wg.Add(len(*sssDB))
	for _, ssDB := range *sssDB {
		go d.dispatchOneProcess(ssDB, serverSessionDao, &wg)
	}
	wg.Wait()
	d.monitor()
	time.Sleep(1 * time.Second)
}

func (d *ServerSessionDispatcher) dispatchOneProcess(ssDB server_session.ServerSession,
	serverSessionDao *server_session.ServerSessionDao, wg *sync.WaitGroup) {
	defer wg.Done()
	// 获取可用process
	dispatchProcess, err := stragegy.Pick(ssDB.FleetID)
	if err != nil {
		go event.RegisterServerSessionError("[server session] dispatch error",
			event.ResourceEvent{ResourceId: ssDB.ID, FleetId: ssDB.FleetID}, err.Error(), event.EventTraceStateIncident)
		log.RunLogger.Errorf("[dispatch] get available process by fleet id "+
			"%s for %s failed because %v", ssDB.FleetID, ssDB.ID, err)
		ssDB.State = common.ServerSessionStateError
		ssDB.StateReason = err.Error()
		_, err := serverSessionDao.Update(&ssDB)
		if err != nil {
			log.RunLogger.Errorf("[dispatch] failed to update error server session %s to db, for %v", ssDB.ID, err)
		}
		return
	}

	log.RunLogger.Infof("[dispatch] success pick process %v for server session %s in fleetID %s",
		dispatchProcess, ssDB.ID, ssDB.FleetID)

	d.doDispatch(dispatchProcess, ssDB)
}

func (d *ServerSessionDispatcher) monitor() {
	d.roundCount += 1
	if d.roundCount >= roundThreshold {
		log.RunLogger.Infof("start to do monitor func and start new round")
		d.roundCount = 0
	}
}

func (d *ServerSessionDispatcher) doDispatch(processId string, ssDB server_session.ServerSession) {
	// 填充数据
	var dispatchProcess app_process.AppProcess
	err := models.MySqlOrm.QueryTable(&app_process.AppProcess{}).Filter("Id", processId).One(&dispatchProcess)
	if err != nil {
		log.RunLogger.Infof("[update dispatch result] get process %s msg from db error %+v", processId, err)
		go event.RegisterServerSessionError("[server session] get process failed",
			event.ResourceEvent{ResourceId: processId, FleetId: ssDB.FleetID}, err.Error(), event.EventTraceStateIncident)
		redisErr := redisserver.DeleteProcessFromRedis(ssDB.FleetID, processId)
		if redisErr != nil {
			log.RunLogger.Errorf("[dispatch] add server session count to redis failed for %v", redisErr)
		}
		return
	}

	ssDB.ProcessID = dispatchProcess.ID
	ssDB.InstanceID = dispatchProcess.InstanceID
	ssDB.PublicIP = dispatchProcess.PublicIP
	ssDB.ClientPort = dispatchProcess.ClientPort
	ssDB.PID = dispatchProcess.PID
	ssDB.State = common.ServerSessionStateActivating
	ssDB.ProtectionPolicy = dispatchProcess.NewServerSessionProtectionPolicy
	ssDB.ProtectionTimeLimitMinutes = dispatchProcess.ServerSessionProtectionTimeLimitMinutes
	ssDB.ActivationTimeoutSeconds = dispatchProcess.ServerSessionActivationTimeoutSeconds

	if err := d.handleDispatch(&ssDB, &dispatchProcess); err != nil {
		log.RunLogger.Errorf("[server session] dispatch server session to process error:%+v", err)
		return
	}

	go func() {
		ss := apis.TransferSSFromModel2Api(&ssDB)
		err = ActivateServerSession(&dispatchProcess, &ssDB, ss, log.RunLogger)
		if err != nil {
			go event.RegisterServerSessionError("[server session] activate failed",
				event.ResourceEvent{ResourceId: ssDB.ID, FleetId: ssDB.FleetID}, err.Error(), event.EventTraceStateIncident)
			log.RunLogger.Errorf("[dispatch] activate server session error %v", err)
			return
		}
	}()
	log.RunLogger.Infof("[dispatch] success dispatch server session %s to process %s in fleetID %s",
		ssDB.ID, dispatchProcess.ID, ssDB.FleetID)
}

func (d *ServerSessionDispatcher) handleDispatch(ssDB *server_session.ServerSession, dispatchProcess *app_process.AppProcess) error {
	// 执行事务
	err := models.DispatchServerSession2Process(ssDB, dispatchProcess)
	if err != nil {
		go event.RegisterServerSessionError("[server session] insert db error",
			event.ResourceEvent{ResourceId: ssDB.ID, FleetId: ssDB.FleetID}, err.Error(), event.EventTraceStateWarning)
		// 落库失败都不会去执行激活，Redis认为当前进程信息不可信，直接删除
		redisErr := redisserver.DeleteProcessFromRedis(ssDB.FleetID, dispatchProcess.ID)
		if redisErr != nil {
			log.RunLogger.Errorf("[dispatch] add server session count to redis failed for %v", redisErr)
		}
		if err.Error() == NoAvailableServerSession {
			log.RunLogger.Infof("[dispatch] current process %s is not active or in max server session", dispatchProcess.ID)
			return err
		}
		log.RunLogger.Errorf("[dispatch] failed to dispatch server session %s to process %s in fleetID %s "+
			"because %v", ssDB.ID, dispatchProcess.ID, ssDB.FleetID, err)
		ssDB.State = common.ServerSessionStateError
		ssDB.StateReason = err.Error()
		serverSessionDao := server_session.NewServerSessionDao(models.MySqlOrm)
		_, err := serverSessionDao.Update(ssDB)
		if err != nil {
			log.RunLogger.Errorf("[dispatch] failed to update error server session %s to db, for %v", ssDB.ID, err)
		}
		log.RunLogger.Infof("[dispatch] dispatch in db failed, begin to add server session to redis")
		return err
	}
	return nil
}
